package com.snopy.mq.activemq;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.lang3.StringUtils;

import javax.jms.*;

/**
 * @author snopy
 * @description:
 * @classname JmsConsumer
 * @date 2020/12/22 9:51
 * @email 77912204@qq.com
 */
public class JmsConsumer {
    //activemq连接地址
    private static final String BROKER_URL = "failover:(tcp://192.168.5.136:61616)?initialReconnectDelay=1000&maxReconnectDelay=30000";

    private static final String mytopic = "mytopic";

    public static void main(String[] args) throws JMSException {
        //1 创建连接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        //2 创建连接
        Connection connection = connectionFactory.createConnection();
        connection.setClientID("067");
        //连接
        connection.start();
        //3 创建会话session
        final Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
        //4 创建消费队列
        //Destination destination = session.createQueue(myqueue);
        Topic topic = session.createTopic(mytopic);
        //5 创建消费者
        //MessageConsumer consumer = session.createConsumer(topic);
        //5创建持久化消费者
        MessageConsumer consumer = session.createDurableSubscriber(topic,"067");
        //6指定消息监听器
        int i = 0;
        consumer.setMessageListener((Message message)-> {
                TextMessage textMessage = (TextMessage) message;
                try {
                    //Thread.sleep(2000);
                    if (StringUtils.isNotEmpty(textMessage.getText())) {
                        System.out.println(textMessage.getText());
                        textMessage.acknowledge();
                        if (textMessage.getText().contains("5")){
                            throw new Exception("错误");
                        }else{
                            session.commit();
                        }

                    }
                } catch (Exception e) {
                    System.out.println(e.getMessage());
                    try {
                        session.rollback();
                    } catch (JMSException e1) {
                        e1.printStackTrace();
                    }
                }

        });

        /*while (true){

        }*/
        //6 消费消息
        //int i = 0;
        /*while (true){

            TextMessage receive = (TextMessage)consumer.receive(6000);

            if (receive!=null){
                try {
                    System.out.println(receive.getText());
                    System.out.println(i/(i-10));
                    receive.acknowledge();
                    session.commit();
                }catch (Exception e){
                    System.out.println("失败了,再来："+e.getMessage());
                    session.recover();
                }
            }else{
                System.out.println(i+"没有消息：");

            }
            i++;
            if(receive==null){
                try {
                    Thread.sleep(1000);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (i>1000){
                    break;
                }
            }else{
                try {
                    System.out.println("休息一下");
                    Thread.sleep(1000);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }        }*/

   /*     consumer.close();
        session.close();
        connection.close();*/
    }
}
